package com.cy.config;

import cn.hutool.json.JSONUtil;
import com.cy.domain.ViolationVehicle;
import com.cy.service.SendMessage;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope;

import java.nio.charset.StandardCharsets;

/**
 * @Author jiangwang
 * @Created by 2021/12/16 9:03
 * @Description: TODO
 */
@Configuration
public class RabbitMqConfig
{

    private final SendMessage sendMessage;

    public RabbitMqConfig(SendMessage sendMessage)
    {
        this.sendMessage = sendMessage;
    }

    @Value("${spring.rabbitmq.host}")
    private String host;

    @Value("${spring.rabbitmq.port}")
    private int port;

    @Value("${spring.rabbitmq.username}")
    private String username;

    @Value("${spring.rabbitmq.password}")
    private String password;

    @Value("${exchange}")
    private String exchange;

    @Value("${queue}")
    private String queue;

    @Value("${routingKey}")
    private String routingKey;

    @Bean
    public ConnectionFactory connectionFactory()
    {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory(host, port);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        connectionFactory.setVirtualHost("/");
        return connectionFactory;
    }

    @Bean
    @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
    /**必须是prototype类型*/
    public RabbitTemplate rabbitTemplate()
    {
        return new RabbitTemplate(connectionFactory());
    }

    /**
     * 针对消费者配置
     * 1. 设置交换机类型
     * 2. 将队列绑定到交换机
     * FanoutExchange: 将消息分发到所有的绑定队列，无routingkey的概念
     * HeadersExchange ：通过添加属性key-value匹配
     * DirectExchange:按照routingkey分发到指定队列
     * TopicExchange:多关键字匹配
     */
    @Bean
    public TopicExchange defaultExchange()
    {
        return new TopicExchange(exchange);
    }

    /**
     * 获取队列A
     *
     * @return
     */
    @Bean
    public Queue queue()
    {
        //队列持久
        return new Queue(queue, true);
    }


    @Bean
    public Binding binding()
    {
        return BindingBuilder.bind(queue()).to(defaultExchange()).with(routingKey);
    }


    @Bean
    public SimpleMessageListenerContainer messageContainer()
    {
        //加载处理消息A的队列
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory());
        //设置接收多个队列里面的消息，这里设置接收队列A
        container.setQueues(queue());
        container.setExposeListenerChannel(true);
        //设置最大的并发的消费者数量
        container.setMaxConcurrentConsumers(1);
        //最小的并发消费者的数量
        container.setConcurrentConsumers(1);
        //设置确认模式手工确认
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        container.setMessageListener(new ChannelAwareMessageListener()
        {
            @Override
            public void onMessage(Message message, Channel channel) throws Exception
            {
                /*通过basic.qos方法设置prefetch_count=1，这样RabbitMQ就会使得每个Consumer在同一个时间点最多处理一个Message，
                 换句话说,在接收到该Consumer的ack前,它不会将新的Message分发给它 */
                channel.basicQos(1);
                String json = new String(message.getBody(), StandardCharsets.UTF_8);
                //进行消息接受处理
                ViolationVehicle vv = JSONUtil.toBean(json, ViolationVehicle.class);
                if (Config.devList.contains(vv.getDeviceNo()) && !"".equals(vv.getPlateNbr()))
                {
                    //发送数据到海康平台
                    sendMessage.sendMessage(vv);
                }
                /*为了保证永远不会丢失消息，RabbitMQ支持消息应答机制。
                 当消费者接收到消息并完成任务后会往RabbitMQ服务器发送一条确认的命令，然后RabbitMQ才会将消息删除。*/
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            }
        });
        return container;
    }
}
